/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce;

import java.io.IOException;
import org.msgpack.MessagePack;
import org.msgpack.packer.BufferPacker;
import org.msgpack.unpacker.BufferUnpacker;
import org.msgpack.MessageTypeException;
import org.sss.mapreduce.datatype.Packable;

public class MessagePackEncoding implements Encoding {
  private MessagePack msgpack = new MessagePack();

  public MessagePackEncoding() {
    this(null);
  }

  public MessagePackEncoding(ClassLoader cl) {
    if (cl != null) {
      msgpack.setClassLoader(cl);
    }
  }

  @Override
  public boolean canEncode(Class<? extends Packable> klass) throws SssException {
    try {
      msgpack.lookup(klass);
      return true;
    }
    catch (MessageTypeException e) {
      return false;
    }
  }

  @Override
  public Encoder createEncoder(Class<? extends Packable> klass) throws SssException {
    assert canEncode(klass): "invalid type";
    final BufferPacker packer = msgpack.createBufferPacker();
    return new Encoder() {
      public byte[] encode(Packable p) throws IOException {
        try {
          packer.write(p);
          return packer.toByteArray();
        }
        finally {
          packer.clear();
        }
      }
    };
  }

  @Override
  public Decoder createDecoder(final Class<? extends Packable> klass) throws SssException {
    assert canEncode(klass): "invalid type";
    final BufferUnpacker unpacker = msgpack.createBufferUnpacker();
    return new Decoder() {
      public Packable decode(byte[] b) throws IOException {
        unpacker.wrap(b);
        return unpacker.read(klass);
      }
    };
  }
}
